package com.sc.passbook.service.impl;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sc.passbook.constant.Constant;
import com.sc.passbook.constant.PassStatus;
import com.sc.passbook.dao.MerchantsDao;
import com.sc.passbook.entity.Merchants;
import com.sc.passbook.mapper.PassMapper;
import com.sc.passbook.service.IUserPassService;
import com.sc.passbook.service.IUserService;
import com.sc.passbook.vo.Pass;
import com.sc.passbook.vo.PassInfo;
import com.sc.passbook.vo.PassTemplate;
import com.sc.passbook.vo.Response;
import com.spring4all.spring.boot.starter.hbase.api.HbaseTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.text.ParseException;
import java.util.Date;
import java.util.List;
import java.util.Map;

import static com.sc.passbook.constant.PassStatus.USED;
import static java.util.stream.Collectors.toList;

/**
 * @author sc
 * Created on  2018/9/2
 */
@Service
@Slf4j
public class UserPassService implements IUserPassService {

    public static final String DATE = "yyyy-MM-dd";
    @Autowired
    private HbaseTemplate hbaseTemplate;
    @Autowired
    private MerchantsDao merchantsDao;

    @Override
    public Response getUserPassInfo(Long userId) throws IOException, ParseException {
        return getPassInfoByStatus(userId,PassStatus.UNUSED);
    }

    @Override
    public Response getUserUsedPassInfo(Long userId) throws IOException, ParseException {
        return getPassInfoByStatus(userId, USED);
    }

    @Override
    public Response getUserAllPassInfo(Long userId) throws IOException, ParseException {
        return getPassInfoByStatus(userId,PassStatus.ALL);

    }

    @Override
    public Response userUsePass(Pass pass) {
        byte[] prefix = Bytes.toBytes(new StringBuilder(String.valueOf(pass.getUserId())).reverse().toString());
        Scan s  = new Scan();
        List<Filter> filters = Lists.newArrayList();
        filters.add(new PrefixFilter(prefix));
        filters.add(new SingleColumnValueFilter(
                Constant.PassTable.FAMILY_I.getBytes(),
                Constant.PassTable.TEMPLATE_ID.getBytes(),
                CompareFilter.CompareOp.EQUAL,
                Bytes.toBytes(pass.getTemplateId())));

        filters.add(new SingleColumnValueFilter(
                Constant.PassTable.FAMILY_I.getBytes(),
                Constant.PassTable.CONSUME_DATE.getBytes(),
                CompareFilter.CompareOp.EQUAL,
                Bytes.toBytes("-1")));
        s.setFilter(new FilterList(filters));
        List<Pass> passes = hbaseTemplate.find(Constant.PassTable.TABLE_NAME, s, new PassMapper());
        if(passes.size()!=1){
            log.error("pass error");
            return Response.fail("pass error");
        }
        byte[] FI = Bytes.toBytes(Constant.PassTable.FAMILY_I);
        byte[] CD = Bytes.toBytes(Constant.PassTable.CONSUME_DATE);
        Put put = new Put(passes.get(0).getRowKey().getBytes());
        put.addColumn(FI,CD,Bytes.toBytes(DateFormatUtils.format(new Date(),"yyyy-MM-dd")));
        hbaseTemplate.saveOrUpdate(Constant.PassTable.TABLE_NAME,put);
        return Response.success();
    }

    private Response getPassInfoByStatus(Long userId, PassStatus status) throws IOException, ParseException {
        byte[] rowPrefix = Bytes.toBytes(new StringBuilder(String.valueOf(userId)).reverse().toString());
        List<Filter> filters = Lists.newArrayList();
        CompareFilter.CompareOp compareOp = status==PassStatus.UNUSED?
                CompareFilter.CompareOp.EQUAL:CompareFilter.CompareOp.NOT_EQUAL;
        Scan scan = new Scan();
        filters.add(new PrefixFilter(rowPrefix));
        if(status!=PassStatus.ALL){
            filters.add(new SingleColumnValueFilter(
                    Constant.PassTable.FAMILY_I.getBytes(),
                    Constant.PassTable.CONSUME_DATE.getBytes(),
                    compareOp,
                    Bytes.toBytes("-1")));
        }
        scan.setFilter(new FilterList(filters));
        List<Pass> passes = hbaseTemplate.find(Constant.PassTable.TABLE_NAME,scan,new PassMapper());
        Map<String, PassTemplate> passTemplateMap = buildPassTemplate(passes);
        Map<Integer, Merchants> merchantsMap = buildMerchantsMap((List<PassTemplate>) passTemplateMap.values());
        List<PassInfo> infos = Lists.newArrayList();

        for (Pass pass : passes) {
            PassTemplate t = passTemplateMap.getOrDefault(pass.getTemplateId(),null);
            if(null==t){
                log.error("passtemplate null ");
                continue;
            }
            Merchants m = merchantsMap.getOrDefault(t.getId(),null);
            if(m==null){
                log.error(" merchants null");
                continue;
            }
            infos.add(new PassInfo(pass,t,m));
        }
        return Response.success(infos);
    }

    private Map<String, PassTemplate> buildPassTemplate(List<Pass> list) throws IOException, ParseException {
        byte[] FAMILY_B = Bytes.toBytes(Constant.PassTemplateTable.FAMILY_B);
        byte[] ID = Bytes.toBytes(Constant.PassTemplateTable.ID);
        byte[] TITLE = Bytes.toBytes(Constant.PassTemplateTable.TITLE);
        byte[] SUMMARY = Bytes.toBytes(Constant.PassTemplateTable.SUMMARY);
        byte[] DESC = Bytes.toBytes(Constant.PassTemplateTable.DESC);
        byte[] HAS_TOKEN = Bytes.toBytes(Constant.PassTemplateTable.HAS_TOKEN);
        byte[] BACKGROUND = Bytes.toBytes(Constant.PassTemplateTable.BACKGROUND);

        byte[] FAMILY_C = Bytes.toBytes(Constant.PassTemplateTable.FAMILY_C);
        byte[] LIMIT = Bytes.toBytes(Constant.PassTemplateTable.LIMIT);
        byte[] START = Bytes.toBytes(Constant.PassTemplateTable.START);
        byte[] END = Bytes.toBytes(Constant.PassTemplateTable.END);

        List<String> ids = list.stream().map(Pass::getTemplateId).collect(toList());
        List<Get> gets = Lists.newArrayListWithExpectedSize(ids.size());
        ids.forEach(i -> gets.add(new Get(Bytes.toBytes(i))));

        Result[] results = hbaseTemplate.getConnection().
                getTable(TableName.valueOf(Constant.PassTemplateTable.TABLE_NAME)).
                get(gets);
        Map<String, PassTemplate> map = Maps.newHashMap();
        for (Result result : results) {
            PassTemplate p = new PassTemplate();
            p.setId(Bytes.toInt(result.getValue(FAMILY_B, ID)));
            p.setTitle(Bytes.toString(result.getValue(FAMILY_B, TITLE)));
            p.setSummary(Bytes.toString(result.getValue(FAMILY_B, SUMMARY)));
            p.setDesc(Bytes.toString(result.getValue(FAMILY_B, DESC)));
            p.setBackground(Bytes.toInt(result.getValue(FAMILY_B, BACKGROUND)));
            p.setHasToken(Bytes.toBoolean(result.getValue(FAMILY_B, HAS_TOKEN)));
            p.setLimit(Bytes.toLong(result.getValue(FAMILY_C, LIMIT)));
            p.setStart(DateUtils.parseDate(Bytes.toString(result.getValue(FAMILY_C, START)), DATE));
            p.setEnd(DateUtils.parseDate(Bytes.toString(result.getValue(FAMILY_C, END)), DATE));
            map.put(Bytes.toString(result.getRow()), p);
        }
        return map;
    }


    private Map<Integer, Merchants> buildMerchantsMap(List<PassTemplate> passTemplates) {
        Map<Integer, Merchants> map = Maps.newHashMap();
        List<Integer> collect = passTemplates.stream().map(PassTemplate::getId).collect(toList());
        List<Merchants> byIdIn = merchantsDao.findByIdIn(collect);
        byIdIn.forEach(i -> map.put(i.getId(), i));
        return map;
    }
}
